package net.bwie.realtime.jtp.dws.log.job;

import net.bwie.realtime.jtp.dws.log.bean.IkAnalyzeFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

/**
 *
 * 当日APP流量日志数据进行实时汇总统计, 搜索关键词的个数
 */

public class JtpTrafficAPPwordMinuteWindowDwsJob {
    public static void main(String[] args) {
        //1.创建执行环境
        TableEnvironment tableEnv = getTableEnv();
        //2.创建输入表
        createInputTable(tableEnv);
        //3.处理数据
        handle(tableEnv);
        //创建输出表
        createOutputTable(tableEnv);
        //保存到clickhouse
//       outputTable.execute().print();
        saveToClickHouse(tableEnv);
        //4.join广告维度表
        joinDim(tableEnv);
    }

    private static void joinDim(TableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery("SELECT\n" +
                "    window_start AS window_start_time,\n" +
                "    window_end AS window_end_time,\n" +
                "    open_ad_id,\n" +
                "    COUNT(open_ad_id) AS open_ad_count,\n" +
                "    SUM(open_ad_ms) AS open_ad_ms,\n" +
                "    SUM(open_ad_skip_ms) AS open_ad_skip_ms,\n" +
                "    UNIX_TIMESTAMP() * 1000 AS ts,\n" +
                "    PROCTIME() AS proc_time\n"+
                "FROM TABLE(\n" +
                "         TUMBLE(TABLE start_log_table,DESCRIPTOR(row_time),INTERVAL '1' MINUTES)\n" +
                " )t1\n" +
                "GROUP BY\n" +
                "    window_start,window_end,\n" +
                "    open_ad_id");
        tableEnv.createTemporaryView("start_ad_result_table", table);

        tableEnv.executeSql("CREATE TABLE `ads_info_mysql_source_dim`\n" +
                "(\n" +
                "  id               INT COMMENT '主键ID',\n" +
                "  name             STRING COMMENT '广告名称',\n" +
                "  content          STRING COMMENT '广告内容（图片URL/视频URL/H5代码）',\n" +
                "  type             STRING COMMENT '广告类型（IMAGE, VIDEO, HTML5）',\n" +
                "  start_time       TIMESTAMP COMMENT '展示开始时间',\n" +
                "  end_time         TIMESTAMP COMMENT '展示结束时间',\n" +
                "  is_active        INT COMMENT '是否启用',\n" +
                "  target_condition STRING COMMENT '定向条件（JSON格式）',\n" +
                "  created_at       TIMESTAMP COMMENT '数据插入时间',\n" +
                "  updated_at       TIMESTAMP COMMENT '数据更新时间',\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'jdbc',\n" +
                "  'url' = 'jdbc:mysql://node103:3306/jtp_ads',\n" +
                "  'table-name' = 'ads_info',\n" +
                "  'lookup.cache.max-rows' = '10',\n" +
                "  'lookup.cache.ttl' = '1 hour',\n" +
                "  'username' = 'root',\n" +
                "  'password' = '123456',\n" +
                "  'driver' = 'com.mysql.cj.jdbc.Driver'\n" +
                ")"
        );
        Table table1 = tableEnv.sqlQuery(
                "SELECT\n" +
                        "    t1.window_start_time\n" +
                        "    , t1.window_end_time\n" +
                        "    , t2.name AS open_ad_name\n" +
                        "    , t1.open_ad_count\n" +
                        "    , t1.open_ad_ms\n" +
                        "    , t1.open_ad_skip_ms\n" +
                        "    , t1.ts\n" +
                        "FROM start_ad_result_table t1\n" +
                        "    LEFT JOIN ads_info_mysql_source_dim FOR SYSTEM_TIME AS OF t1.proc_time AS t2\n" +
                        "        ON t1.open_ad_id =t2.id "
        );
        table1.execute().print();
    }

    private static void createOutputTable(TableEnvironment tableEnv) {
        //APP启动类型统计：设置窗口进行分组、聚合计算
        Table table = tableEnv.sqlQuery(
                        "SELECT\n" +
                        "    TUMBLE_START(row_time, INTERVAL '1' MINUTES) AS window_start_time\n" +
                        "     , TUMBLE_END(row_time, INTERVAL '1' MINUTES) AS window_end_time\n" +
                        "     , entry\n" +
                        "     , count(entry) AS entry_count\n" +
                        "     , sum(loading_time) AS loading_time\n" +
                        "     , UNIX_TIMESTAMP() * 1000 AS ts\n" +
                        "FROM start_log_table\n" +
                        "GROUP BY\n" +
                        "    TUMBLE(row_time, INTERVAL '1' MINUTES ),\n" +
                        "    entry"
        );
        //- s2-APP启动广告统计：设置窗口进行分组、聚合计算
        Table table2 = tableEnv.sqlQuery(
                "SELECT\n" +
                        "    TUMBLE_START(row_time, INTERVAL '1' MINUTE) AS window_start_time,\n" +
                        "    TUMBLE_END(row_time, INTERVAL '1' MINUTE) AS window_end_time,\n" +
                        "    open_ad_id,\n" +
                        "    COUNT(open_ad_id) AS open_ad_count,\n" +
                        "    SUM(open_ad_ms) AS open_ad_ms,\n" +
                        "    SUM(open_ad_skip_ms) AS open_ad_skip_ms,\n" +
                        "    UNIX_TIMESTAMP() * 1000 AS ts\n" +
                        "FROM start_log_table\n" +
                        "GROUP BY\n" +
                        "    TUMBLE(row_time, INTERVAL '1' MINUTE),\n" +
                        "    open_ad_id"
        );
        tableEnv.createTemporaryView("start_report_table",table);
        tableEnv.createTemporaryView("start_ad_report_table",table2);

    }

    private static void saveToClickHouse(TableEnvironment tableEnv) {

        // 创建输出表：映射到Clickhouse表
        //APP启动类型窗口汇总表
        tableEnv.executeSql(
                "CREATE TABLE IF NOT EXISTS dws_traffic_app_start_window_report\n" +
                        "(\n" +
                        "    window_start_time String COMMENT '计算窗口开始日期时间',\n" +
                        "    window_end_time   String COMMENT '计算窗口结束日期时间',\n" +
                        "    entry             String COMMENT '启动app方式，icon手机图标、notice通知、install安装后启动',\n" +
                        "    entry_count       BIGINT COMMENT '启动app方式次数',\n" +
                        "    loading_time      BIGINT COMMENT '启动app加载时间',\n" +
                        "    ts                BIGINT COMMENT '时间戳'\n"+
                        ") WITH (\n"  +
                        "'connector' = 'clickhouse',\n" +
                        "'url' = 'jdbc:clickhouse://node103:8123/jtp_log_report',\n" +
                        "'table' = 'dws_traffic_app_start_window_report',\n" +
                        "'username' = 'default',\n" +
                        "'password' = '',\n" +
                        "'format' = 'json'\n" +
                        ")"
        );
        //APP启动广告窗口汇总表
        tableEnv.executeSql(
                "CREATE TABLE IF NOT EXISTS dws_traffic_app_start_ad_window_report_clickhouse_sink\n" +
                        "(\n" +
                        "    `window_start_time` STRING COMMENT '窗口开始日期时间',\n" +
                        "    `window_end_time`   STRING COMMENT '窗口结束日期时间',\n" +
                        "    `open_ad_id`        INT COMMENT '启动app方式',\n" +
                        "    `open_ad_count`     BIGINT COMMENT '广告展示次数',\n" +
                        "    `open_ad_ms`        BIGINT COMMENT '广告总共播放时间',\n" +
                        "    `open_ad_skip_ms`   BIGINT COMMENT '用户跳过广告时间',\n" +
                        "    `ts`                BIGINT COMMENT '数据计算时间戳'\n" +
                        ") WITH (\n" +
                        "    'connector' = 'clickhouse',\n" +
                        "    'url' = 'jdbc:clickhouse://node103:8123/jtp_log_report',\n" +
                        "    'table' = 'dws_traffic_app_start_ad_window_report',\n" +
                        "    'username' = 'default',\n" +
                        "    'password' = '',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );
        tableEnv.executeSql("INSERT INTO  dws_traffic_app_start_window_report\n"+
                "    SELECT\n" +
                "    DATE_FORMAT(window_start_time,'yyyy-MM-dd HH:mm:ss') AS window_start_time\n" +
                "     ,DATE_FORMAT(window_end_time,'yyyy-MM-dd HH:mm:ss') AS window_end_time\n" +
                "     ,entry\n"+
                "     ,entry_count\n" +
                "     ,loading_time\n" +
                "     ,ts\n" +
                "    FROM start_report_table"
                );
        tableEnv.executeSql("INSERT INTO dws_traffic_app_start_ad_window_report_clickhouse_sink\n" +
                        "SELECT\n" +
                        "    DATE_FORMAT(window_start_time, 'yyyy-MM-dd HH:mm:ss') AS window_start_time\n" +
                        "     , DATE_FORMAT(window_end_time, 'yyyy-MM-dd HH:mm:ss') AS window_end_time\n" +
                        "     , open_ad_id \n" +
                        "     , open_ad_count\n" +
                        "     , open_ad_ms\n" +
                        "     , open_ad_skip_ms\n" +
                        "     , ts\n" +
                        "FROM start_ad_report_table"
                );



    }

    private static void handle(TableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery(
                        "SELECT\n" +
                        "    `start`['entry'] AS entry,\n" +
                        "    CAST(`start`['loading_time'] AS BIGINT) AS loading_time,\n" +
                        "    CAST(`start`['open_ad_id'] AS INT) AS open_ad_id,\n" +
                        "    CAST(`start`['open_ad_ms'] AS BIGINT) AS open_ad_ms,\n" +
                        "    CAST(`start`['open_ad_skip_ms'] AS BIGINT) AS open_ad_skip_ms,\n" +
                        "    row_time\n" +
                        "FROM dwd_log_start_log_kafka_source\n" +
                        "WHERE `start`['entry'] IS NOT NULL"
        );
        tableEnv.createTemporaryView("start_log_table",table);



    }

    private static void createInputTable(TableEnvironment tableEnv) {
        tableEnv.executeSql("CREATE TABLE dwd_log_start_log_kafka_source (\n" +
                "    `common` MAP<STRING, STRING> COMMENT '公共环境信息',\n" +
                "    `start` MAP<STRING, STRING> COMMENT '启动信息',\n" +
                "    `ts` BIGINT COMMENT '时间戳',\n" +
                "    row_time AS TO_TIMESTAMP(FROM_UNIXTIME(ts / 1000, 'yyyy-MM-dd HH:mm:ss.SSS')) COMMENT '事件时间，指定格式',\n" +
                "    WATERMARK FOR row_time AS row_time - INTERVAL '0' MINUTE\n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'dwd-traffic-start-log',\n" +
                "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "    'properties.group.id' = 'gid_dwd_traffic_start_log',\n" +
                "    'scan.startup.mode' = 'earliest-offset',\n" +
                "    'format' = 'json',\n" +
                "    'json.fail-on-missing-field' = 'false',\n" +
                "    'json.ignore-parse-errors' = 'true'\n" +
                ")\n"
        );


    }

    private static TableEnvironment getTableEnv() {
        //  创建TableEnv
        EnvironmentSettings build = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        TableEnvironment tableEnv = TableEnvironment.create(build);
        //设置时区
        Configuration configuration = tableEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        return tableEnv;

    }
}
